package com.wudl.flink.examples;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.CloseableIterator;

/**
 * @ClassName : FlinkMySqlToHbase
 * @Description :
 * @Author :wudl
 * @Date: 2021-08-28 16:52
 */

public class FlinkMySqlToHbase2 {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);



        TableResult inputTable = tableEnv.executeSql("CREATE TABLE  sourceHbaseTable (" +
                "rowkey STRING ," +
                "cf ROW<name STRING > ," +
                "PRIMARY KEY (rowkey) NOT ENFORCED "+
                ") " +
                "WITH (" +
                "'connector' = 'hbase-2.2'," +
                "'table-name' = 'wudl_hbase'," +
                "'sink.buffer-flush.max-rows' = '1000'," +
                "'zookeeper.quorum' = '192.168.1.161:2181'" +
                " )");
        TableResult outPutTable = tableEnv.executeSql("CREATE TABLE  MySqlTable (" +
                "sex STRING ," +
                "name STRING " +

                ") " +
                "WITH (" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://192.168.1.180:3306/MyFlink?useUnicode=true&characterEncoding=UTF-8'," +
                "'table-name' = 'MyHbaseEtl'," +
                " 'username' = 'root'," +
                " 'password' = '123456'" +
                " )");

        Table table = tableEnv.sqlQuery("select sex ,ROW(name) cf  from  MySqlTable");

        tableEnv.executeSql("insert into sourceHbaseTable select *  from "+table);


//        Table table = tableEnv.sqlQuery("SELECT  info.name name ,info.sex sex , cast(info.age as  BIGINT) age FROM sourceHbaseTable");
//        tableEnv.executeSql("insert into MySqlTable select * from "+table);



//        tableEnv.sqlQuery("select * from  MySqlTable").execute().print();
//        Table table = tableEnv.sqlQuery("SELECT rowkey , cf.name FROM sourceHbaseTable");
//
//        // 查询的结果
//        TableResult executeResult = table.execute();
//
//        // 获取查询结果
//        CloseableIterator<Row> collect = executeResult.collect();
//        executeResult.print();
//






    }
}
